package com.yyx.demo.lock;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;

import static org.apache.zookeeper.CreateMode.EPHEMERAL_SEQUENTIAL;
import static org.apache.zookeeper.ZooDefs.Ids.OPEN_ACL_UNSAFE;


public class WriteLock extends ZKLockProtocolSupport{

    private static final Logger LOG = LoggerFactory.getLogger(WriteLock.class);

    private final String dir;
    private ZNodeLockName idName;
    private ZKOperation zop;
    private LockListener callback;
    private byte[] data = {0x12, 0x34};
    private String id;

    //树结构排列id
    private String ownerId;
    //上一个节点id
    private String lastChildId;

    //创建锁类
    public WriteLock(ZooKeeper zookeeper, String dir) {
        super(zookeeper);
        this.dir = dir;
        this.zop = new LockZooKeeperOperation();
    }

    //有锁回调的创建
    public WriteLock(
            ZooKeeper zookeeper,
            String dir,
            List<ACL> acl,
            LockListener callback) {
        this(zookeeper, dir);
        this.callback = callback;
    }


    public synchronized boolean lock() throws KeeperException, InterruptedException {
        //该节点是否处于关闭状态
        if (isClosed()) {
            LOG.debug("节点已经关闭");
            return false;
        }
        //确定路径存在，不存在则创建
        ensurePathExists(dir);
        //尝试锁操作
        return (Boolean) retryOperation(zop);
    }

    public synchronized void unlock() throws RuntimeException {

        if (!isClosed() && id != null) {
            // we don't need to retry this operation in the case of failure
            // as ZK will remove ephemeral files and we don't wanna hang
            // this process when closing if we cannot reconnect to ZK
            try {
                ZKOperation zopdel = () -> {
                    zookeeper.delete(id, -1);
                    return Boolean.TRUE;
                };
                zopdel.execute();
            } catch (InterruptedException e) {
                LOG.warn("Unexpected exception", e);
                // set that we have been interrupted.
                Thread.currentThread().interrupt();
            } catch (KeeperException.NoNodeException e) {
                // do nothing
            } catch (KeeperException e) {
                LOG.warn("Unexpected exception", e);
                throw new RuntimeException(e.getMessage(), e);
            } finally {
                LockListener lockListener = getLockListener();
                if (lockListener != null) {
                    lockListener.lockReleased();
                }
                id = null;
            }
        }
    }

    //锁操作实现类
    private class LockZooKeeperOperation implements ZKOperation {

        private void findPrefixInChildren(String prefix, ZooKeeper zookeeper, String dir)
                throws KeeperException, InterruptedException {

         //看是否以前已经创建过，如果有，则直接使用，没有则创建
            List<String> names = zookeeper.getChildren(dir, false);
            for (String name : names) {
                if (name.startsWith(prefix)) {
                    id = name;
                    LOG.debug("发现已经创建的id: {}", id);
                    break;
                }
            }
            if (id == null) {
                id = zookeeper.create(dir + "/" + prefix, data, OPEN_ACL_UNSAFE, EPHEMERAL_SEQUENTIAL);

                LOG.debug("Created id: {}", id);
            }
        }

        @Override
        public boolean execute() throws KeeperException, InterruptedException {
            do {
                if (id == null) {
                    long sessionId = zookeeper.getSessionId();
                    String prefix = "zk-" + sessionId + "-";

                    findPrefixInChildren(prefix, zookeeper, dir);
                    //通过id创建
                    idName = new ZNodeLockName(id);
                }
                List<String> names = zookeeper.getChildren(dir, false);

                if (names.isEmpty()) {
                    LOG.warn("No children in: {} when we've just created one! Lets recreate it...", dir);
                    // lets force the recreation of the id
                    id = null;
                } else {
                    // lets sort them explicitly (though they do seem to come back in order ususally :)
                    SortedSet<ZNodeLockName> sortedNames = new TreeSet<>();
                    for (String name : names) {
                        sortedNames.add(new ZNodeLockName(dir + "/" + name));
                    }
                    ownerId = sortedNames.first().getName();
                    SortedSet<ZNodeLockName> lessThanMe = sortedNames.headSet(idName);
                    //如果有更小的序列号，子节点watcher它
                    if (!lessThanMe.isEmpty()) {
                        ZNodeLockName lastChildName = lessThanMe.last();
                        lastChildId = lastChildName.getName();
                        LOG.debug("监听比我小的节点: {}", lastChildId);
                        Stat stat = zookeeper.exists(lastChildId, new LockWatcher());

                        if (stat != null) {
                            LOG.debug("先前已有线程拿到锁");
                            return Boolean.FALSE;
                        } else {
                            LOG.warn("没有找到比我小的节点: {}", lastChildName.getName());
                        }
                        //如果没有更小的序列号，建立Listener
                    } else {
                        if (isOwner()) {
                            LockListener lockListener = getLockListener();
                            if (lockListener != null) {
                                lockListener.lockAcquired();
                            }
                            LOG.debug("该线程拿到锁");
                            return Boolean.TRUE;
                        }
                    }
                }
            }
            while (id == null);
            return Boolean.FALSE;
        }

    }

    public synchronized LockListener getLockListener() {
        return this.callback;
    }



    public synchronized void setLockListener(LockListener callback) {
        this.callback = callback;
    }


    public boolean isOwner() {
        return id != null && id.equals(ownerId);
    }

    private class LockWatcher implements Watcher {

        @Override
        public void process(WatchedEvent event) {
            // lets either become the leader or watch the new/updated node
            LOG.debug("监听执行: {}", event);
            LOG.debug("Watcher fired: {}", event);
            try {
                lock();
            } catch (Exception e) {
                LOG.warn("Failed to acquire lock", e);
            }
        }
    }

    public String getDir() {
        return dir;
    }


}
